package com.taurus.permanent.normal;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import com.taurus.core.util.Logger;
import com.taurus.core.util.Utils;
import com.taurus.permanent.TPServer;
import com.taurus.permanent.core.BaseCoreService;
import com.taurus.permanent.core.BitSwarmEngine;
import com.taurus.permanent.core.DefaultConstants;
import com.taurus.permanent.core.ServerConfig;
import com.taurus.permanent.data.IPacketQueue;
import com.taurus.permanent.data.ISocketChannel;
import com.taurus.permanent.data.Packet;
import com.taurus.permanent.data.Session;
import com.taurus.permanent.data.SessionType;
import com.taurus.permanent.io.IOHandler;

/**
 * SocketWriter
 * 
 *
 */
public final class SocketWriter extends BaseCoreService {
	private BitSwarmEngine               engine;
	private IOHandler                    ioHandler;
	private SocketAcceptor               acceptor;
	private final Logger                 logger;
	private final ExecutorService        threadPool;
	private final ExecutorService        udpthreadPool;
	private final BlockingQueue<Session> sessionTicketsQueue;
	private final BlockingQueue<Packet>  udpPacketQueue;

	private int                          udpQueueMaxSize;
	private volatile long                droppedUdpPacketsCount = 0L;
	private volatile int                 tcpThreadId            = 1;
	private volatile int                 udpThreadId            = 1;
	private volatile boolean             isActive               = false;
	private volatile long                droppedPacketsCount    = 0L;
	private volatile long                writtenBytes           = 0L;
	private volatile long                writtenPackets         = 0L;
	private int                          tcpThreadPoolSize;
	private int                          udpThreadPoolSize;

	public SocketWriter() {
		engine = BitSwarmEngine.getInstance();
		this.tcpThreadPoolSize = engine.getConfig().socketWriterThreadPoolSize;
		this.udpThreadPoolSize = engine.getConfig().udpSocketWriterThreadPoolSize;

		threadPool = Executors.newFixedThreadPool(tcpThreadPoolSize);
		udpthreadPool = Executors.newFixedThreadPool(udpThreadPoolSize);
		logger = Logger.getLogger(SocketWriter.class);

		acceptor = engine.getSocketAcceptor();
		sessionTicketsQueue = new LinkedBlockingQueue<Session>();
		udpPacketQueue = new LinkedBlockingQueue<Packet>();

	}

	public void init(Object o) {
		super.init(o);
		if (isActive) {
			throw new IllegalArgumentException("Object is already initialized. Destroy it first!");
		}
		if (tcpThreadPoolSize < 1) {
			throw new IllegalArgumentException("Illegal value for a thread pool size: " + tcpThreadPoolSize);
		}
		engine = BitSwarmEngine.getInstance();
		isActive = true;

		initThreadPool();

		logger.info("Socket Writer started (pool size:" + tcpThreadPoolSize + ")");
	}

	public void destroy(Object o) {
		super.destroy(o);

		isActive = false;
		List<Runnable> leftOvers = threadPool.shutdownNow();
		logger.info("SocketWriter stopped. Unprocessed tasks: " + leftOvers.size());
	}

	public int getQueueSize() {
		return sessionTicketsQueue.size();
	}

	public int getThreadPoolSize() {
		return tcpThreadPoolSize;
	}

	public IOHandler getIOHandler() {
		return ioHandler;
	}

	public void setIOHandler(IOHandler ioHandler) {
		if (this.ioHandler != null) {
			throw new IllegalStateException("You cannot reassign the IOHandler class!");
		}
		this.ioHandler = ioHandler;
	}

	public void continueWriteOp(Session session) {
		if (session != null)
			sessionTicketsQueue.add(session);
	}

	private void initThreadPool() {
		for (int j = 0; j < tcpThreadPoolSize; j++) {
			threadPool.execute(new TCPWriteRunner());
		}

		for (int j = 0; j < udpThreadPoolSize; j++) {
			udpthreadPool.execute(new UDPWriteRunner());
		}
	}

	private void processSessionQueue(ByteBuffer writeBuffer, Session session) {
		if (session != null) {
			SessionType type = session.getType();

			if (type == SessionType.NORMAL) {
				processRegularSession(writeBuffer, session);
			} else if (type == SessionType.VOID)
				return;
		}
	}

	private void processRegularSession(ByteBuffer writeBuffer, Session session) {
		if (session.isFrozen()) {
			return;
		}
		Packet packet = null;
		try {
			IPacketQueue sessionQ = session.getPacketQueue();

			synchronized (sessionQ) {
				if (!sessionQ.isEmpty()) {
					packet = sessionQ.peek();

					if (packet == null) {
						return;
					}
					tcpSend(writeBuffer, sessionQ, session, packet);
				}
			}

		} catch (ClosedChannelException cce) {
			logger.debug("Socket closed during write operation for session: " + session);
		} catch (IOException localIOException) {
		} catch (Exception e) {
			logger.warn("Error during write. Session: " + session);
			logger.error(e);
		}
	}

	private void tcpSend(ByteBuffer writeBuffer, IPacketQueue sessionQ, Session session, Packet packet)
			throws Exception {
		ISocketChannel channel = session.getConnection();
		if (channel == null) {
			logger.debug("Skipping packet, found null socket for Session: " + session);
			return;
		}
		writeBuffer.clear();

		byte[] buffer = packet.isFragmented() ? packet.getFragmentBuffer() : (byte[]) packet.getData();
		if (writeBuffer.capacity() < buffer.length) {
			writeBuffer = Utils.allocateBuffer(buffer.length, engine.getConfig().writeBufferType);
		}
		writeBuffer.put(buffer);
		writeBuffer.flip();

		long toWrite = writeBuffer.remaining();
		long bytesWritten = channel.write(writeBuffer);
		if (bytesWritten == 0)
			throw new IOException("Written Bytes excetion!");

		writtenBytes += bytesWritten;
		session.addWrittenBytes(bytesWritten);
		if (bytesWritten < toWrite) {
			byte[] bb = new byte[writeBuffer.remaining()];
			writeBuffer.get(bb);
			packet.setFragmentBuffer(bb);

			SelectionKey sk = (SelectionKey) session.getSystemProperty(DefaultConstants.SESSION_SELECTION_KEY);
			if ((sk != null) && (sk.isValid())) {
				sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
			} else {
				logger.warn("Could not OP_WRITE for Session: " + session + ", written bytes: " + bytesWritten);
			}
		} else {
			writtenPackets += 1L;

			sessionQ.take();
			if (!sessionQ.isEmpty()) {
				sessionTicketsQueue.add(session);
			}
		}
	}

	private void udpSend(ByteBuffer writeBuffer, byte[] data, Session session) throws Exception {
		writeBuffer.clear();
		if (writeBuffer.capacity() < data.length) {
			writeBuffer = Utils.allocateBuffer(data.length, this.engine.getConfig().writeBufferType);
		}
		writeBuffer.put(data);
		writeBuffer.flip();
		int sourcePort = ((InetSocketAddress) session.getDatagramChannel().getLocalAddress()).getPort();
		DatagramChannel datagramChannel = this.acceptor.getCachedUDPChannel(sourcePort);
		Integer sessionUdpPort = (Integer) session.getSystemProperty("UDPPort");
		if (datagramChannel == null)
			throw new IllegalStateException("UDP Packet cannot be sent to: " + session
					+ ", no bound DatagramChannel available for port: " + sourcePort);
		if (sessionUdpPort == null)
			throw new IllegalStateException("UDP Packet cannot be sent to: " + session + ", no UDP port set.");
		int written = datagramChannel.send(writeBuffer, new InetSocketAddress(session.getAddress(), sessionUdpPort));
		if (written != 0) {
			this.writtenPackets++;
			this.writtenBytes += written;
			session.addWrittenBytes(written);
		} else {
			this.droppedUdpPacketsCount++;
		}
	}

	public void enqueuePacket(Packet packet) {
		if (packet.isUdp()) {
			enqueueUDP(packet);
		} else {
			enqueueTCP(packet);
		}
	}

	private void enqueueTCP(Packet packet) {
		Collection<Session> recipients = packet.getRecipients();
		int size = recipients.size();
		if (recipients != null && size > 0) {
			if (packet.getSender() != null)
				packet.getSender().setLastWriteTime(System.currentTimeMillis());
			if (size == 1) {
				enqueueLocalPacket(packet.getRecipients().iterator().next(), packet);
			} else {
				for (Session session : recipients)
					enqueueLocalPacket(session, packet.clone());
			}
		}
	}

	private void enqueueUDP(Packet packet) {
		Collection<Session> recipients = packet.getRecipients();
		if (recipients != null && recipients.size() > 0) {
			if (packet.getSender() != null)
				packet.getSender().setLastWriteTime(System.currentTimeMillis());
			if (this.udpPacketQueue.size() <= this.udpQueueMaxSize) {
				this.udpPacketQueue.add(packet);
			} else {
				this.droppedUdpPacketsCount++;
			}
		}
	}

	private void enqueueLocalPacket(Session session, Packet packet) {
		IPacketQueue sessionQ = session.getPacketQueue();

		if (sessionQ != null) {
			synchronized (sessionQ) {
				try {
					boolean wasEmpty = sessionQ.isEmpty();

					sessionQ.put(packet);

					if ((wasEmpty)) {
						sessionTicketsQueue.add(session);
					}

					packet.setRecipients(null);
				} catch (Exception error) {
					dropOneMessage(session);
				}
			}
		}
	}

	/**
	 * 丢包处理
	 * 
	 * @param session
	 */
	private void dropOneMessage(Session session) {
		session.addDroppedMessages(1);
		droppedPacketsCount += 1L;
	}

	public long getDroppedPacketsCount() {
		return droppedPacketsCount;
	}

	public long getDroppedUdpPacketCount() {
		return this.droppedUdpPacketsCount;
	}

	public long getWrittenBytes() {
		return writtenBytes;
	}

	public long getWrittenPackets() {
		return writtenPackets;
	}

	private class TCPWriteRunner implements Runnable {
		private TCPWriteRunner() {
		}

		public void run() {
			Thread.currentThread().setName("SocketWriter[TCP]-" + tcpThreadId++);

			ServerConfig setting = TPServer.me().getConfig();
			ByteBuffer writeBuffer = Utils.allocateBuffer(setting.maxWriteBufferSize, setting.writeBufferType);

			while (isActive) {
				try {
					Session session = sessionTicketsQueue.take();
					processSessionQueue(writeBuffer, session);
				} catch (InterruptedException e) {
					logger.warn("SocketWriter thread interrupted: " + Thread.currentThread());
					isActive = false;
				} catch (Throwable t) {
					logger.warn("Problems in SocketWriter main loop, Thread: " + Thread.currentThread());
					logger.error(t);
				}
			}

			logger.info("SocketWriter threadpool shutting down.");
		}
	}

	private class UDPWriteRunner implements Runnable {
		private UDPWriteRunner() {
		}

		public void run() {
			Thread.currentThread().setName("SocketWriter[UDP]-" + udpThreadId++);

			ServerConfig setting = TPServer.me().getConfig();
			ByteBuffer writeBuffer = Utils.allocateBuffer(setting.maxWriteBufferSize, setting.writeBufferType);
			while (isActive) {
				try {
					Packet udpPacket = udpPacketQueue.take();
					byte[] data = null;
					for (Session session : udpPacket.getRecipients()) {
						try {
							data = (byte[]) udpPacket.getData();

							udpSend(writeBuffer, data, session);
						} catch (Exception ioex) {
							ioex.printStackTrace();
							logger.warn("Problems in SocketWriter::UDP inner loop.", ioex);
							droppedUdpPacketsCount++;
						}
					}
				} catch (InterruptedException ex) {
					logger.warn("SocketWriter::UDP thread interrupted", ex);
					isActive = false;
				} catch (Throwable ex) {
					logger.warn("Problems in SocketWriter::UDP main loop.", ex);
				}
			}
			logger.info("SocketWriter::UDP threadpool shutting down.");
		}
	}
}
